// 实现异步工作器
#ifndef __M_LOPPER_H__
#define __M_LOPPER_H__

#include "buffer.hpp"
#include <mutex>
#include <condition_variable>
#include <thread>
#include <functional>
#include <memory>
#include <atomic>

namespace log
{
    using Functor = std::function<void(Buffer &)>;
    enum class AsyncType
    {
        ASYNC_SAFE, // 安全状态,表示缓冲区满了则阻塞,避免资源耗尽风险
        ASUNC_UNSAFE
    };
    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(const Functor &cb, AsyncType looper_type = AsyncType::ASYNC_SAFE) : _stop(false),
                                                                                        _thread(std::thread(&AsyncLooper::threadEntry, this)),
                                                                                        _callBack(cb),
                                                                                        _looper_type(looper_type) {}
        ~AsyncLooper()
        {
            stop();
        }
        void stop()
        {
            _stop = true;           // 将退出标志设置为true
            _cond_con.notify_all(); // 唤醒所有工作线程去将消费缓冲区的数据进行落地
            _thread.join();         // 等待工作线程退出,代表数据已经处理完了
        }
        void push(const char *data, size_t len)
        {
            // 1. 无线扩容-非安全
            std::unique_lock<std::mutex> lock(_mutex);
            // 阻塞条件:后面的表达式为假,则会被阻塞在该语句
            if (_looper_type == AsyncType::ASYNC_SAFE)
                _cond_pro.wait(lock, [&]()
                               { return _pro_buf.writeAbleSize() >= len; }); // 若生产缓冲区剩余空间大小大于等于len,则可以添加数据
            // 走到这里说明可以向该缓冲区添加数据
            _pro_buf.push(data, len);
            _cond_con.notify_all(); // 上面添加数据后就可以让在该消费条件变量等待下的消费者对缓冲区的数据进行处理,实现同步
        }

    private:
        // 线程入口函数-对消费缓冲区中的数据进行处理，处理完毕后，初始化缓冲区，交换缓冲区
        void threadEntry()
        {
            while (1)
            {
                // 1. 判断生产缓冲区有无数据，有责交换,无则阻塞
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _cond_con.wait(lock, [&]()
                                   { return _stop || !_pro_buf.empty(); }); // 有数据处理，停止工作器也要唤醒
                    //工作器停止的同时生产缓冲区为空,这是在跳出.否则会造成生产缓冲区中有数据，但是没有被完全处理
                    if(_stop&&_pro_buf.empty()) break;
                    _con_buf.swap(_pro_buf);                                // 这就是实现将生产缓冲区的数据"拷贝"至消费缓冲区,而这里使用一个挪动两个缓冲区指针的想法,减少拷贝
                    // 2. 唤醒生产者
                    if (_looper_type == AsyncType::ASYNC_SAFE)
                        _cond_pro.notify_all();
                }
                // 交换完缓冲区后,这是就可以释放锁了,否则导致其它生产者就无法向生产缓冲区添加数据,因为要申请锁,而前面的线程没有释放锁,此时只能等
                // 3. 被唤醒后，对消费缓冲区进行数据处理
                _callBack(_con_buf); // 使用回调函数对消费缓冲区的数据进行处理
                // 4. 初始化缓冲区
                _con_buf.reset();
            }
        }

    private:
        Functor _callBack;

    private:
        AsyncType _looper_type;
        std::atomic<bool> _stop;           // 工作器停止标志,false表示不停止,true表示停止
        Buffer _pro_buf;                   // 生产缓冲区
        Buffer _con_buf;                   // 消费缓冲区
        std::mutex _mutex;                 // 互斥锁
        std::condition_variable _cond_pro; //
        std::condition_variable _cond_con;
        std::thread _thread; // 异步工作器对应的工作线程
    };
}

#endif